//! 查询执行引擎性能基准测试
//! 
//! 本模块包含对查询执行引擎各个算子的性能基准测试，
//! 用于验证执行引擎的性能表现并与SQLite进行对比。

use std::time::Instant;
use std::sync::Arc;
use crate::storage::{StorageEngine, page::PageManager, DataType};
use crate::config::DatabaseConfig;
use crate::sql::{Value, execution::{
    ExecutionContext, ExecutionSchema, ColumnInfo, Operator,
    ExecutionTuple, VersionInfo,
    operators::{
        TableScanOperator, ProjectionOperator, FilterOperator,
        AggregationOperator, SortOperator,
        SortKey, SortDirection, AggregateExpression, AggregateFunction,
        FilterPredicate, ComparisonOperator
    }
}};
use crate::Result;

/// 性能基准测试结果
#[derive(Debug, Clone)]
pub struct BenchmarkResult {
    /// 测试名称
    pub name: String,
    /// 执行时间（微秒）
    pub execution_time_us: u64,
    /// 处理的行数
    pub rows_processed: usize,
    /// 输出的行数
    pub rows_output: usize,
    /// 吞吐量（行/秒）
    pub throughput_rows_per_sec: f64,
    /// 内存使用量（字节）
    pub memory_usage_bytes: usize,
}

impl BenchmarkResult {
    /// 创建新的基准测试结果
    pub fn new(
        name: String,
        execution_time_us: u64,
        rows_processed: usize,
        rows_output: usize,
        memory_usage_bytes: usize,
    ) -> Self {
        let throughput_rows_per_sec = if execution_time_us > 0 {
            (rows_processed as f64) / (execution_time_us as f64 / 1_000_000.0)
        } else {
            0.0
        };

        Self {
            name,
            execution_time_us,
            rows_processed,
            rows_output,
            throughput_rows_per_sec,
            memory_usage_bytes,
        }
    }

    /// 打印基准测试结果
    pub fn print(&self) {
        println!("=== 基准测试结果: {} ===", self.name);
        println!("执行时间: {:.2} ms", self.execution_time_us as f64 / 1000.0);
        println!("处理行数: {}", self.rows_processed);
        println!("输出行数: {}", self.rows_output);
        println!("吞吐量: {:.2} 行/秒", self.throughput_rows_per_sec);
        println!("内存使用: {:.2} KB", self.memory_usage_bytes as f64 / 1024.0);
        println!();
    }
}

/// 性能基准测试套件
pub struct ExecutionBenchmarks {
    /// 存储引擎
    storage_engine: StorageEngine,
    /// 执行上下文
    context: ExecutionContext,
}

impl ExecutionBenchmarks {
    /// 将Vec<Vec<Value>>转换为Vec<ExecutionTuple>
    fn convert_to_execution_tuples(&self, data: Vec<Vec<Value>>) -> Vec<ExecutionTuple> {
        data.into_iter()
            .enumerate()
            .map(|(i, values)| ExecutionTuple {
                values,
                row_id: Some(crate::storage::tuple::RowId::new(0, i as u16)),
                version_info: Some(VersionInfo {
                    xmin: 1,
                    xmax: None,
                    committed: true,
                }),
            })
            .collect()
    }

    /// 创建新的基准测试套件
    pub fn new() -> Result<Self> {
        let config = DatabaseConfig::default();
        let storage_engine = StorageEngine::new(&config)?;

        // 创建页面管理器
        let page_manager = Arc::new(PageManager::new(1000)); // 1000页缓存
        let context = ExecutionContext::new(page_manager, 1);

        Ok(Self {
            storage_engine,
            context,
        })
    }

    /// 生成测试数据
    fn generate_test_data(&self, rows: usize) -> Vec<Vec<Value>> {
        let mut data = Vec::with_capacity(rows);
        
        for i in 0..rows {
            data.push(vec![
                Value::Integer(i as i64),
                Value::Text(format!("name_{}", i)),
                Value::Real((i as f64) * 1.5),
                Value::Boolean(i % 2 == 0),
                Value::Integer((i % 10) as i64), // category
            ]);
        }
        
        data
    }



    /// 基准测试：表扫描算子
    pub fn benchmark_table_scan(&mut self, rows: usize) -> Result<BenchmarkResult> {
        // 创建测试模式
        let schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("value".to_string(), DataType::Real, false),
            ColumnInfo::new("active".to_string(), DataType::Boolean, false),
            ColumnInfo::new("category".to_string(), DataType::Integer, false),
        ]);

        // 创建表扫描算子
        let mut scan_op = TableScanOperator::new(
            "test_table".to_string(),
            None, // 无投影，扫描所有列
            &self.context
        )?;

        // 设置测试数据
        let test_data = self.convert_to_execution_tuples(self.generate_test_data(rows));
        scan_op.set_test_data(test_data);

        // 执行基准测试
        let start_time = Instant::now();
        scan_op.open(&mut self.context)?;

        let mut output_rows = 0;
        while let Some(_tuple) = scan_op.next(&mut self.context)? {
            output_rows += 1;
        }

        scan_op.close(&mut self.context)?;
        let execution_time = start_time.elapsed();

        Ok(BenchmarkResult::new(
            format!("表扫描 ({} 行)", rows),
            execution_time.as_micros() as u64,
            rows,
            output_rows,
            rows * 5 * 8, // 估算内存使用量
        ))
    }

    /// 基准测试：投影算子
    pub fn benchmark_projection(&mut self, rows: usize) -> Result<BenchmarkResult> {
        // 创建输入表扫描
        let schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("value".to_string(), DataType::Real, false),
            ColumnInfo::new("active".to_string(), DataType::Boolean, false),
            ColumnInfo::new("category".to_string(), DataType::Integer, false),
        ]);

        let mut input_scan = TableScanOperator::new(
            "test_table".to_string(),
            None,
            &self.context
        )?;
        input_scan.set_test_data(self.convert_to_execution_tuples(self.generate_test_data(rows)));

        // 创建投影算子（选择前3列）
        let projection_indices = vec![0, 1, 2];

        let mut projection_op = ProjectionOperator::new(
            Box::new(input_scan),
            projection_indices,
        )?;

        // 执行基准测试
        let start_time = Instant::now();
        projection_op.open(&mut self.context)?;

        let mut output_rows = 0;
        while let Some(_tuple) = projection_op.next(&mut self.context)? {
            output_rows += 1;
        }

        projection_op.close(&mut self.context)?;
        let execution_time = start_time.elapsed();

        Ok(BenchmarkResult::new(
            format!("投影 ({} 行)", rows),
            execution_time.as_micros() as u64,
            rows,
            output_rows,
            rows * 3 * 8, // 估算内存使用量
        ))
    }

    /// 基准测试：过滤算子
    pub fn benchmark_filter(&mut self, rows: usize) -> Result<BenchmarkResult> {
        // 创建输入表扫描
        let schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("value".to_string(), DataType::Real, false),
            ColumnInfo::new("active".to_string(), DataType::Boolean, false),
            ColumnInfo::new("category".to_string(), DataType::Integer, false),
        ]);

        let mut input_scan = TableScanOperator::new("test_table".to_string(), None, &self.context)?;
        input_scan.set_test_data(self.convert_to_execution_tuples(self.generate_test_data(rows)));

        // 创建简单的过滤谓词（id >= 0，基本上总是返回true）
        let predicate = FilterPredicate::Comparison {
            left_column: 0,
            operator: ComparisonOperator::GreaterThanOrEqual,
            right_value: Value::Integer(0),
        };

        // 创建过滤算子
        let mut filter_op = FilterOperator::new(
            Box::new(input_scan),
            predicate,
        );

        // 执行基准测试
        let start_time = Instant::now();
        filter_op.open(&mut self.context)?;

        let mut output_rows = 0;
        while let Some(_tuple) = filter_op.next(&mut self.context)? {
            output_rows += 1;
        }

        filter_op.close(&mut self.context)?;
        let execution_time = start_time.elapsed();

        Ok(BenchmarkResult::new(
            format!("过滤 ({} 行)", rows),
            execution_time.as_micros() as u64,
            rows,
            output_rows,
            rows * 5 * 8, // 估算内存使用量
        ))
    }

    /// 基准测试：聚合算子
    pub fn benchmark_aggregation(&mut self, rows: usize) -> Result<BenchmarkResult> {
        // 创建输入表扫描
        let schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("value".to_string(), DataType::Real, false),
            ColumnInfo::new("active".to_string(), DataType::Boolean, false),
            ColumnInfo::new("category".to_string(), DataType::Integer, false),
        ]);

        let mut input_scan = TableScanOperator::new("test_table".to_string(), None, &self.context)?;
        input_scan.set_test_data(self.convert_to_execution_tuples(self.generate_test_data(rows)));

        // 创建聚合算子（按category分组，计算count和avg）
        let group_by_columns = vec![4]; // category列
        let aggregates = vec![
            AggregateExpression {
                function: AggregateFunction::Count,
                column_index: None,
                output_name: "count".to_string(),
            },
            AggregateExpression {
                function: AggregateFunction::Avg,
                column_index: Some(2), // value列
                output_name: "avg_value".to_string(),
            },
        ];

        let output_schema = ExecutionSchema::new(vec![
            ColumnInfo::new("category".to_string(), DataType::Integer, false),
            ColumnInfo::new("count".to_string(), DataType::Integer, false),
            ColumnInfo::new("avg_value".to_string(), DataType::Real, false),
        ]);

        let mut agg_op = AggregationOperator::new(
            Box::new(input_scan),
            group_by_columns,
            aggregates,
            output_schema,
        );

        // 执行基准测试
        let start_time = Instant::now();
        agg_op.open(&mut self.context)?;

        let mut output_rows = 0;
        while let Some(_tuple) = agg_op.next(&mut self.context)? {
            output_rows += 1;
        }

        agg_op.close(&mut self.context)?;
        let execution_time = start_time.elapsed();

        Ok(BenchmarkResult::new(
            format!("聚合 ({} 行)", rows),
            execution_time.as_micros() as u64,
            rows,
            output_rows,
            rows * 5 * 8 + output_rows * 3 * 8, // 估算内存使用量
        ))
    }

    /// 基准测试：排序算子
    pub fn benchmark_sort(&mut self, rows: usize) -> Result<BenchmarkResult> {
        // 创建输入表扫描
        let schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("value".to_string(), DataType::Real, false),
            ColumnInfo::new("active".to_string(), DataType::Boolean, false),
            ColumnInfo::new("category".to_string(), DataType::Integer, false),
        ]);

        let mut input_scan = TableScanOperator::new("test_table".to_string(), None, &self.context)?;
        input_scan.set_test_data(self.convert_to_execution_tuples(self.generate_test_data(rows)));

        // 创建排序算子（按value降序排序）
        let sort_keys = vec![
            SortKey {
                column_index: 2, // value列
                direction: SortDirection::Descending,
            }
        ];

        let mut sort_op = SortOperator::new(
            Box::new(input_scan),
            sort_keys,
            schema,
        );

        // 执行基准测试
        let start_time = Instant::now();
        sort_op.open(&mut self.context)?;

        let mut output_rows = 0;
        while let Some(_tuple) = sort_op.next(&mut self.context)? {
            output_rows += 1;
        }

        sort_op.close(&mut self.context)?;
        let execution_time = start_time.elapsed();

        Ok(BenchmarkResult::new(
            format!("排序 ({} 行)", rows),
            execution_time.as_micros() as u64,
            rows,
            output_rows,
            rows * 5 * 8 * 2, // 排序需要额外内存
        ))
    }

    /// 运行所有基准测试
    pub fn run_all_benchmarks(&mut self) -> Result<Vec<BenchmarkResult>> {
        let test_sizes = vec![1000, 10000, 100000];
        let mut results = Vec::new();

        for &size in &test_sizes {
            println!("运行基准测试，数据量: {} 行", size);
            
            // 表扫描基准测试
            let scan_result = self.benchmark_table_scan(size)?;
            scan_result.print();
            results.push(scan_result);

            // 投影基准测试
            let proj_result = self.benchmark_projection(size)?;
            proj_result.print();
            results.push(proj_result);

            // 过滤基准测试
            let filter_result = self.benchmark_filter(size)?;
            filter_result.print();
            results.push(filter_result);

            // 聚合基准测试
            let agg_result = self.benchmark_aggregation(size)?;
            agg_result.print();
            results.push(agg_result);

            // 排序基准测试
            let sort_result = self.benchmark_sort(size)?;
            sort_result.print();
            results.push(sort_result);

            println!("--- 完成 {} 行数据的基准测试 ---\n", size);
        }

        Ok(results)
    }

    /// 压力测试：大数据集处理
    pub fn stress_test_large_dataset(&mut self) -> Result<Vec<BenchmarkResult>> {
        println!("🔥 开始大数据集压力测试...");
        let mut results = Vec::new();

        // 测试不同规模的数据集
        let large_sizes = vec![500_000, 1_000_000, 2_000_000];

        for &size in &large_sizes {
            println!("测试数据量: {} 行", size);

            // 大数据集表扫描
            let start = Instant::now();
            let scan_result = self.benchmark_table_scan(size)?;
            let scan_time = start.elapsed();

            println!("  表扫描: {:.2}ms, 吞吐量: {:.0} 行/秒",
                scan_time.as_millis(),
                size as f64 / scan_time.as_secs_f64()
            );
            results.push(scan_result);

            // 大数据集过滤
            let start = Instant::now();
            let filter_result = self.benchmark_filter(size)?;
            let filter_time = start.elapsed();

            println!("  过滤操作: {:.2}ms, 吞吐量: {:.0} 行/秒",
                filter_time.as_millis(),
                size as f64 / filter_time.as_secs_f64()
            );
            results.push(filter_result);

            // 大数据集聚合
            let start = Instant::now();
            let agg_result = self.benchmark_aggregation(size)?;
            let agg_time = start.elapsed();

            println!("  聚合操作: {:.2}ms, 吞吐量: {:.0} 行/秒",
                agg_time.as_millis(),
                size as f64 / agg_time.as_secs_f64()
            );
            results.push(agg_result);

            println!("--- 完成 {} 行数据的压力测试 ---\n", size);
        }

        Ok(results)
    }

    /// 压力测试：复杂查询
    pub fn stress_test_complex_queries(&mut self) -> Result<Vec<BenchmarkResult>> {
        println!("🔥 开始复杂查询压力测试...");
        let mut results = Vec::new();
        let data_size = 100_000;

        // 测试复杂的多算子组合查询
        for i in 1..=5 {
            println!("复杂查询测试 #{}", i);

            let start = Instant::now();
            let result = self.benchmark_complex_query(data_size, i)?;
            let query_time = start.elapsed();

            println!("  查询 #{}: {:.2}ms, 输出行数: {}",
                i, query_time.as_millis(), result.rows_output
            );
            results.push(result);
        }

        Ok(results)
    }

    /// 基准测试：复杂查询（多算子组合）
    fn benchmark_complex_query(&mut self, rows: usize, query_type: usize) -> Result<BenchmarkResult> {
        // 创建测试模式
        let schema = ExecutionSchema::new(vec![
            ColumnInfo::new("id".to_string(), DataType::Integer, false),
            ColumnInfo::new("name".to_string(), DataType::Text(Some(50)), false),
            ColumnInfo::new("value".to_string(), DataType::Real, false),
            ColumnInfo::new("active".to_string(), DataType::Boolean, false),
            ColumnInfo::new("category".to_string(), DataType::Integer, false),
        ]);

        let test_data = self.generate_test_data(rows);

        let start_time = Instant::now();
        let mut output_rows = 0;

        match query_type {
            1 => {
                // 查询1: 扫描 -> 过滤 -> 投影
                let mut scan_op = TableScanOperator::new("test_table".to_string(), None, &self.context)?;
                scan_op.set_test_data(self.convert_to_execution_tuples(test_data));

                let predicate = FilterPredicate::Comparison {
                    left_column: 0,
                    operator: ComparisonOperator::GreaterThan,
                    right_value: Value::Integer(rows as i64 / 2),
                };
                let mut filter_op = FilterOperator::new(Box::new(scan_op), predicate);

                let mut proj_op = ProjectionOperator::new(
                    Box::new(filter_op),
                    vec![0, 1, 2]
                )?;

                proj_op.open(&mut self.context)?;
                while let Some(_tuple) = proj_op.next(&mut self.context)? {
                    output_rows += 1;
                }
                proj_op.close(&mut self.context)?;
            },
            2 => {
                // 查询2: 扫描 -> 聚合 -> 排序
                let mut scan_op = TableScanOperator::new("test_table".to_string(), None, &self.context)?;
                scan_op.set_test_data(self.convert_to_execution_tuples(test_data));

                let aggregates = vec![
                    AggregateExpression {
                        function: AggregateFunction::Count,
                        column_index: None,
                        output_name: "count".to_string(),
                    },
                    AggregateExpression {
                        function: AggregateFunction::Avg,
                        column_index: Some(2),
                        output_name: "avg_value".to_string(),
                    },
                ];

                // 创建聚合输出模式
                let agg_schema = ExecutionSchema::new(vec![
                    ColumnInfo::new("category".to_string(), DataType::Integer, false),
                    ColumnInfo::new("count".to_string(), DataType::Integer, false),
                    ColumnInfo::new("avg_value".to_string(), DataType::Real, false),
                ]);

                let mut agg_op = AggregationOperator::new(
                    Box::new(scan_op),
                    vec![4], // 按category分组
                    aggregates,
                    agg_schema,
                );

                agg_op.open(&mut self.context)?;
                while let Some(_tuple) = agg_op.next(&mut self.context)? {
                    output_rows += 1;
                }
                agg_op.close(&mut self.context)?;
            },
            _ => {
                // 默认查询: 简单扫描
                let mut scan_op = TableScanOperator::new("test_table".to_string(), None, &self.context)?;
                scan_op.set_test_data(self.convert_to_execution_tuples(test_data));

                scan_op.open(&mut self.context)?;
                while let Some(_tuple) = scan_op.next(&mut self.context)? {
                    output_rows += 1;
                }
                scan_op.close(&mut self.context)?;
            }
        }

        let execution_time = start_time.elapsed();

        Ok(BenchmarkResult::new(
            format!("复杂查询 #{} ({} 行)", query_type, rows),
            execution_time.as_micros() as u64,
            rows,
            output_rows,
            rows * 5 * 8,
        ))
    }

    /// 压力测试：并发操作模拟
    pub fn stress_test_concurrent_operations(&mut self) -> Result<Vec<BenchmarkResult>> {
        println!("🔥 开始并发操作压力测试...");
        let mut results = Vec::new();

        // 模拟多个并发查询
        for concurrent_level in vec![1, 2, 4, 8] {
            println!("并发级别: {} 个查询", concurrent_level);

            let start = Instant::now();
            let mut total_output = 0;

            for _i in 0..concurrent_level {
                let result = self.benchmark_table_scan(50_000)?;
                total_output += result.rows_output;
            }

            let total_time = start.elapsed();

            println!("  总时间: {:.2}ms, 总输出行数: {}, 平均吞吐量: {:.0} 行/秒",
                total_time.as_millis(),
                total_output,
                total_output as f64 / total_time.as_secs_f64()
            );

            results.push(BenchmarkResult::new(
                format!("并发测试 ({}个查询)", concurrent_level),
                total_time.as_micros() as u64,
                50_000 * concurrent_level,
                total_output,
                50_000 * concurrent_level * 5 * 8,
            ));
        }

        Ok(results)
    }

    /// 运行完整的压力测试套件
    pub fn run_stress_tests(&mut self) -> Result<()> {
        println!("🚀 开始CDB v3.0压力测试套件...\n");

        // 大数据集压力测试
        let _large_results = self.stress_test_large_dataset()?;

        // 复杂查询压力测试
        let _complex_results = self.stress_test_complex_queries()?;

        // 并发操作压力测试
        let _concurrent_results = self.stress_test_concurrent_operations()?;

        println!("✅ 所有压力测试完成！");
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_benchmark_suite() {
        let mut benchmarks = ExecutionBenchmarks::new().unwrap();
        
        // 运行小规模基准测试
        let scan_result = benchmarks.benchmark_table_scan(1000).unwrap();
        assert!(scan_result.execution_time_us > 0);
        assert_eq!(scan_result.rows_processed, 1000);
        assert_eq!(scan_result.rows_output, 1000);
        
        let proj_result = benchmarks.benchmark_projection(1000).unwrap();
        assert!(proj_result.execution_time_us > 0);
        assert_eq!(proj_result.rows_processed, 1000);
        assert_eq!(proj_result.rows_output, 1000);
    }
}
